package io.kiki.sba.registry.server.data_store.slot;

import io.kiki.sba.registry.api.exchange.RequestException;
import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.PublisherDigestUtil;
import io.kiki.sba.registry.common.model.RegisterVersion;
import io.kiki.sba.registry.common.model.dataserver.DatumDigest;
import io.kiki.sba.registry.common.model.dataserver.DatumSummary;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffDigestRequest;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffDigestResult;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffPublisherRequest;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffPublisherResult;
import io.kiki.sba.registry.common.model.slot.filter.SyncSlotAcceptorManager;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.StringInterner;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.change.DataChangeType;
import io.kiki.sba.registry.server.data_store.lease.ClientInterfaceLeaseManager;
import io.kiki.sba.registry.server.data_store.multi.cluster.slot.MultiClusterSlotMetrics.RemoteSyncLeader;
import io.kiki.sba.registry.server.data_store.pubiterator.DatumBiConsumer;
import io.kiki.sba.registry.server.shared.remoting.ClientSideExchanger;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.Map.Entry;

import static io.kiki.sba.registry.server.data_store.slot.SlotMetrics.SyncLeader;
import static io.kiki.sba.registry.server.data_store.slot.SlotMetrics.SyncSession;


public final class SlotDiffSyncer {
    private final Logger logger = LoggerFactory.getLogger(SlotDiffSyncer.class);
    private final ServerConfig serverConfig;

    @Getter
    private final DatumStorageDelegate datumStorageDelegate;
    private final DataChangeEventCenter dataChangeEventCenter;
    private final ClientInterfaceLeaseManager clientInterfaceLeaseManager;

    private final SyncSlotAcceptorManager syncSlotAcceptorManager;

    public SlotDiffSyncer(ServerConfig serverConfig, DatumStorageDelegate datumStorageDelegate, DataChangeEventCenter dataChangeEventCenter, ClientInterfaceLeaseManager clientInterfaceLeaseManager, SyncSlotAcceptorManager syncSlotAcceptorManager) {
        this.serverConfig = serverConfig;
        this.datumStorageDelegate = datumStorageDelegate;
        this.dataChangeEventCenter = dataChangeEventCenter;
        this.clientInterfaceLeaseManager = clientInterfaceLeaseManager;
        this.syncSlotAcceptorManager = syncSlotAcceptorManager;

    }

    static Map<String, DatumSummary> getSummaryForSyncPublishers(DataSlotDiffDigestResult dataSlotDiffDigestResult, Map<String, DatumSummary> digestSummaryMap) {
        final Map<String, DatumSummary> map = new HashMap<>(dataSlotDiffDigestResult.getUpdateAndAddSize());
        for (String add : dataSlotDiffDigestResult.getAddedDataInfoIds()) {
            map.put(add, new DatumSummary(add));
        }
        for (String update : dataSlotDiffDigestResult.getUpdatedDataInfoIds()) {
            map.put(update, digestSummaryMap.get(update));
        }
        return map;
    }

    static Map<String, DatumSummary> pickSummaries(Map<String, DatumSummary> syncSummaries, int n) {
        Map<String, DatumSummary> map = new HashMap<>();
        for (Map.Entry<String, DatumSummary> entry : syncSummaries.entrySet()) {
            // at least pick one
            map.put(entry.getKey(), entry.getValue());
            int versionSize = entry.getValue().getRegisterIdToRegisterVersionMap().size();
            if (versionSize == 0) {
                // is empty, budget=1
                versionSize = 1;
            }
            n -= versionSize;
            if (n <= 0) {
                break;
            }
        }
        return map;
    }

    DataSlotDiffPublisherResult processSyncPublisherResp(boolean syncLocal, String syncDataCenter, int slotId, GenericResponse<DataSlotDiffPublisherResult> dataSlotDiffPublisherResultGenericResponse, String targetAddress, Map<String, DatumSummary> summaryMap) {
        if (dataSlotDiffPublisherResultGenericResponse == null || !dataSlotDiffPublisherResultGenericResponse.isSuccess()) {
            logger.error("DiffPublisherFailed, syncLocal={}, syncDataCenter={}, slotId={} from {}, resp={}", syncLocal, syncDataCenter, slotId, targetAddress, dataSlotDiffPublisherResultGenericResponse);
            return null;
        }
        DataSlotDiffPublisherResult dataSlotDiffPublisherResult = dataSlotDiffPublisherResultGenericResponse.getData();
        final String slotIdStr = String.valueOf(slotId);

        // sync from session
        final ClientInterfaceId clientInterfaceId = dataSlotDiffPublisherResult.getClientInterfaceId();
        if (syncLocal && clientInterfaceId != null) {
            clientInterfaceLeaseManager.renewal(clientInterfaceId);
        }

        if (dataSlotDiffPublisherResult.isEmpty()) {
            logger.info("DiffPublisherEmpty, syncLocal={}, syncDataCenter={}, slotId={} from {}", syncLocal, syncDataCenter, slotId, targetAddress);
            return dataSlotDiffPublisherResult;
        }

        final Set<String> changeDataIds = new HashSet<>();
        for (Map.Entry<String, List<Publisher>> entry : dataSlotDiffPublisherResult.getUpdatedPublishers().entrySet()) {
            // cache the dataInfoId as key
            final String dataInfoId = StringInterner.intern(entry.getKey());
            final List<Publisher> publisherList = entry.getValue();
            Publisher.internPublisher(publisherList);
            DatumVersion datumVersion = datumStorageDelegate.putPublisher(syncDataCenter, dataInfoId, publisherList);
            if (datumVersion != null) {
                changeDataIds.add(dataInfoId);
            }
            if (!syncLocal) {
                for (Publisher publisher : publisherList) {
                    logger.info("pub,{},{},{},{},{},{},{}", syncDataCenter, slotIdStr, publisher.getDataInfoId(), publisher.getRegisterId(), publisher.getVersion(), publisher.getRegisterTimestamp(), datumVersion);
                }
            }
        }
        // for sync publishers
        for (Map.Entry<String, List<String>> listEntry : dataSlotDiffPublisherResult.getRemovedPublishers().entrySet()) {
            final String dataInfoId = listEntry.getKey();
            final List<String> registerIdList = listEntry.getValue();
            final DatumSummary datumSummary = summaryMap.get(dataInfoId);
            if (datumSummary == null) {
                throw new IllegalArgumentException(StringFormatter.format("dataCenter: {} sync publisher with not exist local DatumSummary: {}", syncDataCenter, dataInfoId));
            }
            Map<String, RegisterVersion> versionMap = datumSummary.getRegisterIdToRegisterVersionMap(registerIdList);
            DatumVersion datumVersion = datumStorageDelegate.removePublishers(syncDataCenter, clientInterfaceId, dataInfoId, versionMap);
            if (datumVersion != null) {
                changeDataIds.add(dataInfoId);
            }

            if (!syncLocal) {
                for (Entry<String, RegisterVersion> entry : versionMap.entrySet()) {
                    logger.info("unpub,{},{},{},{},{},{},{}", syncDataCenter, slotIdStr, dataInfoId, entry.getKey(), entry.getValue().getVersion(), entry.getValue().getRegisterTimestamp(), datumVersion);
                }
            }
        }

        triggerDataChange(syncLocal, syncDataCenter, clientInterfaceId, changeDataIds);
        logger.info("DiffPublisher, synLocal={}, syncDataCenter={}, slotId={} from {}, updatedP {}:{}, removedP {}:{}", syncLocal ? 'Y' : 'N', syncDataCenter, slotId, targetAddress, dataSlotDiffPublisherResult.getUpdatedPublishers().size(), dataSlotDiffPublisherResult.getUpdatedPublishersCount(), dataSlotDiffPublisherResult.getRemovedPublishersCount(), dataSlotDiffPublisherResult.getRemovedPublishers().size());
        return dataSlotDiffPublisherResult;
    }

    boolean syncPublishers(String localDataCenter, boolean syncLocal, String syncDataCenter, int slotId, String targetAddress, ClientSideExchanger exchanger, long slotsEpoch, Map<String, DatumSummary> summaryMap, int maxPublishers, SyncContinues continues, boolean syncSession) {
        // need the empty dataInfoId to add updatePublisher
        Map<String, DatumSummary> round = pickSummaries(summaryMap, maxPublishers);
        // sync for the existing dataInfoIds.publisher
        while (!summaryMap.isEmpty()) {
            if (!continues.continues()) {
                logger.info("syncing publishers break, slotId={} from {}", slotId, targetAddress);
                return true;
            }
            // maybe to many publishers, spit round
            int pubSize = DatumSummary.countPublisherSize(round.values());
            if (syncSession) {
                SyncSession.observeSyncSessionPub(slotId, pubSize);
            } else
                if (syncLocal) {
                    SyncLeader.observeSyncLeaderPub(slotId, pubSize);
                } else {
                    RemoteSyncLeader.observeSyncLeaderPub(syncDataCenter, slotId, pubSize);
                }

            DataSlotDiffPublisherRequest request = new DataSlotDiffPublisherRequest(localDataCenter, slotsEpoch, slotId, syncSlotAcceptorManager, round.values());

            GenericResponse<DataSlotDiffPublisherResult> resp = (GenericResponse<DataSlotDiffPublisherResult>) exchanger.requestRaw(targetAddress, request).getResult();
            DataSlotDiffPublisherResult result = processSyncPublisherResp(syncLocal, syncDataCenter, slotId, resp, targetAddress, round);
            if (result == null) {
                return false;
            }
            if (!result.isHasRemain()) {
                // the sync round has finish, enter next round
                round.keySet().forEach(d -> summaryMap.remove(d));
                round = pickSummaries(summaryMap, maxPublishers);
            } else {
                // has remain, remove the synced dataInfoIds, enter next round
                Set<String> synced = result.syncDataInfoIds();
                for (String dataInfoId : synced) {
                    round.remove(dataInfoId);
                    summaryMap.remove(dataInfoId);
                }
            }
        }
        return true;
    }

    boolean sync(String localDataCenter, String syncDataCenter, boolean syncLocal, int slotId, String targetAddress, long slotLeaderEpoch, ClientSideExchanger exchanger, long slotsEpoch, String summaryTargetIp, int maxPublishers, SyncContinues continues, Map<String, DatumSummary> summaryMap) {
        final boolean syncSession = summaryTargetIp != null;
        if (syncSession) {
            SyncSession.observeSyncSessionId(slotId, summaryMap.size());
        } else
            if (syncLocal) {
                SyncLeader.observeSyncLeaderId(slotId, summaryMap.size());
            } else {
                RemoteSyncLeader.observeSyncLeaderId(syncDataCenter, slotId, summaryMap.size());
            }
        Map<String, DatumDigest> digestMap = PublisherDigestUtil.digest(summaryMap);
        DataSlotDiffDigestRequest request = DataSlotDiffDigestRequest.buildRequest(localDataCenter, slotsEpoch, slotId, slotLeaderEpoch, digestMap, syncSlotAcceptorManager);
        Response exchangeResp = exchanger.requestRaw(targetAddress, request);
        GenericResponse<DataSlotDiffDigestResult> resp = (GenericResponse<DataSlotDiffDigestResult>) exchangeResp.getResult();
        DataSlotDiffDigestResult dataSlotDiffDigestResult = processSyncDigestResp(syncLocal, syncDataCenter, slotId, resp, targetAddress, summaryMap);
        if (dataSlotDiffDigestResult == null) {
            return false;
        }
        if (dataSlotDiffDigestResult.getUpdateAndAddSize() == 0) {
            // no change of update and add
            return true;
        }
        final Map<String, DatumSummary> newSummaryMap = getSummaryForSyncPublishers(dataSlotDiffDigestResult, summaryMap);
        return syncPublishers(localDataCenter, syncLocal, syncDataCenter, slotId, targetAddress, exchanger, slotsEpoch, newSummaryMap, maxPublishers, continues, syncSession);
    }

    DataSlotDiffDigestResult processSyncDigestResp(boolean syncLocal, String syncDataCenter, int slotId, GenericResponse<DataSlotDiffDigestResult> resp, String targetAddress, Map<String, DatumSummary> summaryMap) {
        if (resp == null || !resp.isSuccess()) {
            logger.error("DiffDigestFailed, syncLocal={}, syncDataCenter={}, slotId={} from {}, resp={}", syncLocal, syncDataCenter, slotId, targetAddress, resp);
            return null;
        }
        DataSlotDiffDigestResult dataSlotDiffDigestResult = resp.getData();

        // sync from session
        final ClientInterfaceId clientInterfaceId = dataSlotDiffDigestResult.getClientInterfaceId();
        if (clientInterfaceId != null) {
            clientInterfaceLeaseManager.renewal(clientInterfaceId);
        }

        if (dataSlotDiffDigestResult.isEmpty()) {
            logger.info("DiffDigestEmpty, syncLocal={}, syncDataCenter={}, slotId={} from {}", syncLocal, syncDataCenter, slotId, targetAddress);
            return dataSlotDiffDigestResult;
        }
        // do nothing with added dataInfoId, the added publishers would sync and update by
        // sync.publisher
        // if we create a new empty datum when absent, it's dangerous.
        // if some not expect error occurs when sync publisher and no publisher write to datum,
        // it maybe trigger a empty push with bigger datum.version which created by new empty
        final Set<String> changeDataIds = new HashSet<>();
        for (String removeDataInfoId : dataSlotDiffDigestResult.getRemovedDataInfoIds()) {
            if (datumStorageDelegate.removePublishers(syncDataCenter, clientInterfaceId, removeDataInfoId, summaryMap.get(removeDataInfoId).getRegisterIdToRegisterVersionMap()) != null) {
                changeDataIds.add(removeDataInfoId);
            }
        }

        triggerDataChange(syncLocal, syncDataCenter, clientInterfaceId, changeDataIds);
        logger.info("DiffDigest, syncLocal={}, syncDataCenter={}, slotId={} from {}, update={}, add={}, remove={}, adds={}, removes={}", syncLocal ? 'Y' : 'N', syncDataCenter, slotId, targetAddress, dataSlotDiffDigestResult.getUpdatedDataInfoIds().size(), dataSlotDiffDigestResult.getAddedDataInfoIds().size(), dataSlotDiffDigestResult.getRemovedDataInfoIds().size(), dataSlotDiffDigestResult.getAddedDataInfoIds(), dataSlotDiffDigestResult.getRemovedDataInfoIds());
        return dataSlotDiffDigestResult;
    }

    private void triggerDataChange(boolean syncLocal, String syncDataCenter, ClientInterfaceId clientInterfaceId, Set<String> changeDataIds) {
        if (needTrigger(syncLocal, clientInterfaceId) && !changeDataIds.isEmpty()) {
            dataChangeEventCenter.onChange(changeDataIds, syncLocal ? DataChangeType.sync : DataChangeType.sync_remote, syncDataCenter);
        }
    }

    /**
     * need to trigger change: 1.sync remote data; 2.sync local session;
     *
     * @param syncLocal
     * @param clientInterfaceId
     * @return
     */
    private boolean needTrigger(boolean syncLocal, ClientInterfaceId clientInterfaceId) {
        if (!syncLocal) {
            // sync remote, need to trigger change
            return true;
        }
        // sync local session, need to trigger change
        return clientInterfaceId != null;
    }

    /**
     * summary == null means can not assembly summary at first(migrating); do
     * getDatumSummary(sessionIp)
     *
     * @param slotId          slotId
     * @param sessionIp       sessionIp
     * @param exchanger       exchanger
     * @param slotsEpoch      slotsEpoch
     * @param continues       continues
     * @param summary         summary
     * @param slotLeaderEpoch slotLeaderEpoch
     * @param syncDataCenter  syncDataCenter
     * @return boolean
     * @throws RequestException RequestException
     */
    public boolean syncSession(String syncDataCenter, int slotId, String sessionIp, long slotLeaderEpoch, ClientSideExchanger exchanger, long slotsEpoch, SyncContinues continues, Map<String, DatumSummary> summary) throws RequestException {
        StringUtil.checkNotBlank(sessionIp, "sessionIp");

        // summary == null means can not assembly summary before(eg:migrating);
        // can not change to CollectionUtils.isEmpty
        if (summary == null) {
            // SingletonMap unsupported computeIfAbsent
            final Map<String, Map<String, DatumSummary>> datumSummary = new HashMap<>(1);
            datumSummary.put(sessionIp, new HashMap<>(64));

            datumStorageDelegate.foreach(serverConfig.getDataCenter(), slotId, DatumBiConsumer.publisherGroupsBiConsumer(datumSummary, Collections.singleton(sessionIp), syncSlotAcceptorManager));
            summary = datumSummary.get(sessionIp);
        }

        return sync(syncDataCenter, syncDataCenter, true, slotId, sessionIp, slotLeaderEpoch, exchanger, slotsEpoch, sessionIp, serverConfig.getSlotSyncPublisherDigestMaxNum(), continues, summary);
    }

    public boolean syncSlotLeader(String localDataCenter, String syncDataCenter, boolean syncLocal, int slotId, String slotLeaderIp, long slotLeaderEpoch, ClientSideExchanger exchanger, long slotsEpoch, SyncContinues continues) throws RequestException {
        StringUtil.checkNotBlank(slotLeaderIp, "slotLeaderIp");
        Map<String, DatumSummary> summaries = new HashMap<>();
        datumStorageDelegate.foreach(syncDataCenter, slotId, DatumBiConsumer.publisherGroupsBiConsumer(summaries, syncSlotAcceptorManager));
        return sync(localDataCenter, syncDataCenter, syncLocal, slotId, slotLeaderIp, slotLeaderEpoch, exchanger, slotsEpoch, null, serverConfig.getSlotSyncPublisherDigestMaxNum(), continues, summaries);
    }


}
